package com.xiao.dynamic_rule.functions;

import com.xiao.dynamic_rule.pojo.LogBean;
import com.xiao.dynamic_rule.pojo.ResultBean;
import com.xiao.dynamic_rule.pojo.RuleParam;
import com.xiao.dynamic_rule.pojo.RuleParamAtomic;
import com.xiao.dynamic_rule.service.*;
import com.xiao.dynamic_rule.utils.RuleSimulator;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.ArrayList;
import java.util.List;

/**
 * 规则处理主逻辑
 */
public class RuleProcessFunctionV2 extends KeyedProcessFunction<String, LogBean, ResultBean> {
    // 创建解析规则的类
    private UserProfileQueryService userProfileQueryService;
    private UserActionCountQueryService userActionCountQueryServiceState;
    private UserActionSequenceQueryService userActionSequenceQueryServiceState;

    private UserActionCountQueryService userActionCountQueryServiceClickhouse;
    private UserActionSequenceQueryService userActionSequenceQueryServiceClickhouse;
    private RuleParam ruleParam;
    private ListState<LogBean> listState;

    /**
     * 创建解析规则实现类
     *
     * @param parameters
     * @throws Exception
     */
    @Override
    public void open(Configuration parameters) throws Exception {
        userProfileQueryService = new UserProfileQueryServiceHbaseImpl();

        //构建底层核心state查询服务
        userActionCountQueryServiceState = new UserActionCountQueryServiceStateImpl();
        userActionSequenceQueryServiceState = new UserActionSequenceQueryServiceStateImpl();

        //构建底层核心clickhouse查询服务
        userActionCountQueryServiceClickhouse = new UserActionCountQueryServiceClickHouseImpl();
        userActionSequenceQueryServiceClickhouse = new UserActionSequenceQueryServiceClickhouseImpl();

        /**
         * 获取规则参数
         * TODO 模拟器生成的规则，后面要改成传入
         */
        ruleParam = RuleSimulator.getRuleParam();

        //创建一个存储数据的 listState
        ListStateDescriptor<LogBean> desc = new ListStateDescriptor<>("eventState", LogBean.class);
        //开启state的生命周期,存活俩小时
        StateTtlConfig ttlConfig = StateTtlConfig.newBuilder(Time.hours(2)).updateTtlOnCreateAndWrite().build();
        desc.enableTimeToLive(ttlConfig);
        listState = getRuntimeContext().getListState(desc);

    }

    /**
     * 2小时内的条件在state中查，2小时外的去clickhouse中查
     * @param logBean 传过来的封装成对象的数据
     * @param context flink上下文
     * @param out process的收集器，将处理的数据发往下游
     * @throws Exception
     */
    @Override
    public void processElement(LogBean logBean, Context context, Collector<ResultBean> out) throws Exception {

        //将数据放入state中
        listState.add(logBean);

        //算出切分查询方式的分界点时间
        long splitPoint = System.currentTimeMillis() - 2 * 60 * 60 * 1000;

        //判断是否满足触发条件
        if (ruleParam.getTriggerParam().getEventId().equals(logBean.getEventId())) {

            System.out.println("满足触发条件，即将查询画像，触发条件为：" + ruleParam.getTriggerParam().getEventId() + "deviceID= " + logBean.getDeviceId());

            //画像条件
            boolean profileMatch = userProfileQueryService.judgeQueryProfileCondition(logBean.getDeviceId(), ruleParam);
            if (!profileMatch) return;

            System.out.println("满足画像条件，即将查询行为次数条件");

            //遍历行为次数条件，按照时间跨度，分为两类
            ArrayList<RuleParamAtomic> farRangParams = new ArrayList<>();
            ArrayList<RuleParamAtomic> nearRangParams = new ArrayList<>();

            List<RuleParamAtomic> userActionCountParams = ruleParam.getUserActionCountParams();
            for (RuleParamAtomic userActionCountParam : userActionCountParams) {

                if (userActionCountParam.getRangeStart() < splitPoint){
                    farRangParams.add(userActionCountParam);
                }else {
                    nearRangParams.add(userActionCountParam);
                }
            }

            //行为次数条件
            //两小时内，查询state
            if (nearRangParams.size() > 0) {
                ruleParam.setUserActionCountParams(nearRangParams);
                boolean countMatch = userActionCountQueryServiceState.queryActionCounts("", listState, ruleParam);
                if (!countMatch) return;
            }
            //两小时外，查询clickhouse
            if (farRangParams.size() > 0) {
                ruleParam.setUserActionCountParams(farRangParams);
                boolean countMatch = userActionCountQueryServiceClickhouse.queryActionCounts(logBean.getDeviceId(), null, ruleParam);
                if (!countMatch) return;
            }

            System.out.println("满足行为次数条件，即将查询行为序列条件");

            //行为序列条件
            //行为序列条件整个所有条件只有一个时间范围，所以随便取一个条件的起始时间就可以判断
            //序列条件中的起始时间 < 分界时间，就到clickhouse中查
            if (ruleParam.getUserActionSequenceParams() != null && ruleParam.getUserActionSequenceParams().get(0).getRangeStart() < splitPoint){
                boolean sequenceMatch = userActionSequenceQueryServiceClickhouse.queryActionSequence(logBean.getDeviceId(), null, ruleParam);
                if (!sequenceMatch) return;
            }
            //序列条件中的起始时间 > 分界时间，就到state中查
            if (ruleParam.getUserActionSequenceParams() != null && ruleParam.getUserActionSequenceParams().get(0).getRangeStart() >= splitPoint){
                boolean sequenceMatch = userActionSequenceQueryServiceState.queryActionSequence("", listState, ruleParam);
                if (!sequenceMatch) return;
            }
            // 说明都满足，返回成功结果集
            out.collect(new ResultBean(ruleParam.getRuleId(), logBean.getDeviceId(), logBean.getTimeStamp()));
        }


    }
}
